package com.platform.system.mqtt.utils;

import com.platform.common.constant.Constants;
import com.platform.system.domain.dto.device.DeviceInfoDto;
import com.platform.system.enums.TopicEnums;
import com.platform.system.logic.DeviceInfoLogic;
import com.platform.system.mqtt.event.DeviceListenerEvent;
import com.platform.system.service.ISysDeviceInfoService;
import com.platform.system.service.ISysUpgradeTaskService;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author Huangkai
 */
@Slf4j
@Service("mqttService")
public class MqttServiceImpl implements MqttService {

    @Resource
    DeviceInfoLogic deviceInfoLogic;

    @Resource
    ISysUpgradeTaskService sysUpgradeTaskService;

    @Resource
    ApplicationEventPublisher applicationEventPublisher;


    @Resource
    ISysDeviceInfoService sysDeviceInfoService;

    private final Map<String, ChannelHandlerContext> clients = new ConcurrentHashMap<>();


    @Override
    public void processConnect(ChannelHandlerContext ctx, MqttConnectMessage msg) {
        String clientId = msg.payload().clientIdentifier();
        clients.put(clientId, ctx);
        DeviceInfoDto deviceInfoDto = sysDeviceInfoService.selectDeviceInfoDtoByKey(clientId);
        if(deviceInfoDto != null){
            applicationEventPublisher.publishEvent(new DeviceListenerEvent(this, Constants.NUM_ONE, deviceInfoDto));
            log.info("客户端id{}", clientId);
        }
        // 返回CONNACK消息
        MqttConnAckMessage connAckMessage = MqttMessageBuilders.connAck()
                .returnCode(MqttConnectReturnCode.CONNECTION_ACCEPTED)
                .sessionPresent(false)
                .build();
        ctx.writeAndFlush(connAckMessage);

    }

    @Override
    public void processPublish(ChannelHandlerContext ctx, MqttPublishMessage msg) {
        String topic = msg.variableHeader().topicName();
        ByteBuf payload = msg.payload();
        byte[] headBytes = new byte[payload.readableBytes()];
        payload.readBytes(headBytes);
        String content = new String(headBytes, StandardCharsets.UTF_8);
        System.out.println("Received message on topic " + topic + ": " + content);
    }

    @Override
    public void processSubscribe(ChannelHandlerContext ctx, MqttSubscribeMessage msg) {
        String topic ="";
        // 处理订阅逻辑
        List<MqttTopicSubscription> subscriptions = msg.payload().topicSubscriptions();
        for (MqttTopicSubscription subscription : subscriptions) {
            topic = subscription.topicName();
        }
        // 返回SUBACK消息
        MqttSubAckMessage subAckMessage = MqttMessageBuilders.subAck()
                .packetId(msg.variableHeader().messageId())
                .addGrantedQos(MqttQoS.AT_MOST_ONCE)
                .build();
        ctx.writeAndFlush(subAckMessage);

        log.info("主题{}", topic);
        log.info("是否订阅获取设备信息{}", topic.equals(TopicEnums.DEVICE_INFO.getTopic()));
        if(topic.equals(TopicEnums.DEVICE_INFO.getTopic())){
            //获取客户段id
            String clientId = getClientId(ctx);
            DeviceInfoDto deviceInfoDto = sysDeviceInfoService.selectDeviceInfoDtoByKey(clientId);
            if(deviceInfoDto !=  null){
                deviceInfoLogic.publishDeviceInfo(deviceInfoDto);
            }
        }
    }

    @Override
    public void processUnsubscribe(ChannelHandlerContext ctx, MqttUnsubscribeMessage msg) {
        // 处理取消订阅逻辑
        for (String topic : msg.payload().topics()) {
            System.out.println("Client unsubscribed from: " + topic);
        }
        // 返回UNSUBACK消息
        MqttUnsubAckMessage unsubAckMessage = MqttMessageBuilders.unsubAck()
                .packetId(msg.variableHeader().messageId())
                .build();
        ctx.writeAndFlush(unsubAckMessage);
    }

    @Override
    public void processDisconnect(ChannelHandlerContext ctx) {
        // 处理断开连接逻辑
        String clientId = getClientId(ctx);
        if (clientId != null) {
            clients.remove(clientId);
            DeviceInfoDto deviceInfoDto = sysDeviceInfoService.selectDeviceInfoDtoByKey(clientId);
            log.info("mqtt断开连接查询的设备信息{}", deviceInfoDto);
            if(deviceInfoDto != null){
                applicationEventPublisher.publishEvent(new DeviceListenerEvent(this, Constants.NUM_ZERO, deviceInfoDto));
                System.out.println("Client disconnected: " + clientId);
            }
        }
        ctx.close();
    }

    @Override
    public void processPingReq(ChannelHandlerContext ctx) {
        // 返回PINGRESP消息
        log.info("心跳消息");
        ctx.writeAndFlush(MqttMessage.PINGRESP);
    }

    @Override
    public void publishToClient(List<String> clientIdList, String topic, String message) {
        for(String clientId : clientIdList){
            ChannelHandlerContext ctx = clients.get(clientId);
            if (ctx != null) {
                MqttPublishMessage publishMessage = MqttMessageBuilders.publish()
                        .topicName(topic)
                        .retained(false)
                        .qos(MqttQoS.AT_MOST_ONCE)
                        .payload(Unpooled.copiedBuffer(message, StandardCharsets.UTF_8))
                        .build();
                ctx.writeAndFlush(publishMessage);
            }
        }
    }

    private String getClientId(ChannelHandlerContext ctx) {
        for (Map.Entry<String, ChannelHandlerContext> entry : clients.entrySet()) {
            if (entry.getValue().equals(ctx)) {
                return entry.getKey();
            }
        }
        return null;
    }
}
